FLINK-38647 reports a NullPointerException from the Postgres Pipeline source
when a captured table has REPLICA IDENTITY DEFAULT and an UPDATE arrives that
does not change the primary key. Root cause: PostgresDataSource hardcodes
DebeziumChangelogMode.ALL, which makes the deserializer extract a before-image
that is null under DEFAULT replication, NPE'ing in
DebeziumSchemaDataTypeInference.inferStruct.

This change fixes the issue from two angles:

1. Connector side — expose 'changelog-mode' YAML option on the Postgres
   Pipeline connector (mirrors the legacy SQL connector). Accepts 'all'
   (default, current behaviour) or 'upsert'. In upsert mode the source emits
   UPDATE events with before == null and only after populated, so the
   pipeline runs cleanly under REPLICA IDENTITY DEFAULT without requiring
   FULL (which roughly multiplies WAL volume per UPDATE by column count).

2. Runtime side — make DataChangeEventSerializer null-tolerant. Three changes:
   - copy() null-guards each recordDataSerializer.copy() call so the chained
     CopyingChainingOutput.pushToOperator path stops NPE'ing on null
     before/after.
   - serialize() writes 2 leading boolean presence flags before conditionally
     writing each record. Required because the previous serialize() already
     skipped writing null fields, but deserialize() always tried to read two
     records back — the wire format itself couldn't roundtrip a null-before
     UPDATE.
   - deserialize() reads the 2 flag bytes and skips the corresponding read
     when absent.

Wire format change is symmetric but breaking vs current 3.6.x. Production
checkpoints/savepoints written by older versions cannot be restored by the
new code without a snapshot-version-aware deserialize path; see the PR
description for the proposed compat strategy (bumping
DataChangeEventSerializerSnapshot.CURRENT_VERSION to 2 and reading old
format when version == 1). Happy to add that in a follow-up commit on this
PR — wanted reviewers' opinion on the strategy first.

Signed-off-by: Mehmet Can Şakiroğlu <cansakiroglu@gmail.com>

…n, versioned deserialize, tests

- PostgresDataSourceOptions: define 'changelog-mode' via
  enumType(DebeziumChangelogMode.class), aligning with the SQL connector;
  invalid values now fail at option validation.
- PostgresDataSourceFactory: drop the string switch; validate that all
  captured tables have a primary key when changelog-mode=upsert, mirroring
  PostgreSQLTableFactory.
- DataChangeEventSerializer: introduce CURRENT_VERSION = 2 and a
  version-aware deserialize(int, DataInputView) that still reads the
  pre-flags op-dependent layout, following the
  PhysicalColumnSerializer/SchemaSerializer convention. DataChangeEvent is
  never persisted in checkpoint state, so the snapshot stays a
  SimpleTypeSerializerSnapshot; the versioned overload keeps the old layout
  readable if event bytes are ever embedded in versioned state.
- Tests: null-before UPDATE events in DataChangeEventSerializerTest test
  data (also exercised via EventSerializerTest), legacy-version and
  unknown-version deserialize tests, an upsert-mode roundtrip test, and
  factory tests for option parsing and PK validation.

… null-before UPDATE events (#1)

An UPDATE event with a null before image breaks the implicit contract
that runtime operators and sinks rely on. Instead of making them
null-tolerant, the source now emits REPLACE events (which carry the
after image only) in upsert changelog mode. REPLACE is already handled
by the event serializer and all pipeline sinks, so the null-tolerant
DataChangeEventSerializer changes are reverted and no sink changes are
needed.